package net.qiqbframework.messaging.task;

import net.qiqbframework.common.transaction.Transaction;
import net.qiqbframework.common.transaction.TransactionManager;
import net.qiqbframework.messaging.Message;
import net.qiqbframework.messaging.MetaData;
import net.qiqbframework.messaging.ResultMessage;
import net.qiqbframework.messaging.correlation.CorrelationDataProvider;

import javax.annotation.Nonnull;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * 将每个消息的处理过程封装为任务
 *
 * @param <T> 消息类型
 */
public interface MessageTask<T extends Message<?>> {
    /**
     * @return 当前任务阶段
     */
    TaskPhase phase();


    /**
     * 开始阶段
     *
     *
     *
     *
     *
     *
     *
     */
    void start();

    /**
     * 规划
     */
    void plan();

    void onPlan(Consumer<MessageTask<T>> handler);

    /**
     * 执行
     * @param task
     * @param rollbackConfiguration
     * @return
     * @param <R>
     */
    <R> ResultMessage<R> executeWithResult(Callable<R> task, @Nonnull RollbackConfiguration rollbackConfiguration);

    /**
     * 提交阶段
     */
    void commit();

    void onPrepareCommit(Consumer<MessageTask<T>> handler);

    /**
     *
     */
    void onCommit(Consumer<MessageTask<T>> handler);


    void afterCommit(Consumer<MessageTask<T>> handler);

    /**
     * 异常回滚
     * @param cause
     */
    void rollback(Throwable cause);

    void onRollback(Consumer<MessageTask<T>> handler);

    default boolean isActive() {
        return phase().isStarted();
    }

    void onCleanup(Consumer<MessageTask<T>> handler);


    Optional<MessageTask<?>> parent();

    default boolean isRoot() {
        return parent().isEmpty();
    }


    default MessageTask<?> root() {
        //
        final Optional<MessageTask<?>> parent = parent();
        if (parent.isPresent()){
            return parent.get().root();
        }
        return this;
    }

    T getMessage();

    MessageTask<T> transformMessage(Function<T, ? extends Message<?>> transformOperator);


    MetaData getCorrelationData();

    Map<String, Object> resources();

    @SuppressWarnings("unchecked")
    default <R> R getResource(String name) {
        return (R) resources().get(name);
    }

    @SuppressWarnings("unchecked")
    default <R> R getOrComputeResource(String key, Function<? super String, R> mappingFunction) {
        return (R) resources().computeIfAbsent(key, mappingFunction);
    }

    @SuppressWarnings("unchecked")
    default <R> R getOrDefaultResource(String key, R defaultValue) {
        return (R) resources().getOrDefault(key, defaultValue);
    }

    void registerCorrelationDataProvider(CorrelationDataProvider correlationDataProvider);


    default void attachTransaction(TransactionManager transactionManager) {
        try {
            Transaction transaction = transactionManager.startTransaction();
            onCommit(u -> {
                transaction.commit();
            });
            onRollback(u -> transaction.rollback());
        } catch (Throwable t) {
            rollback(t);
            throw t;
        }
    }



    ExecutionResult getExecutionResult();


    boolean isRolledBack();


    default boolean isCurrent() {
        return CurrentMessageTask.isStarted() && CurrentMessageTask.get() == this;
    }

}
